package com.example.rocketmqdemo.transaction.infr.transactionexecutor;

import com.example.rocketmqdemo.transaction.domain.gateway.LocalTransactionRecordRepository;
import com.example.rocketmqdemo.transaction.domain.transactionexecutor.LocalTransactionExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Component
@RocketMQTransactionListener
public class MRocketMQLocalTransactionListener implements RocketMQLocalTransactionListener {

    @Autowired
    private LocalTransactionRecordRepository localTransactionRecordRepository;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        if (arg instanceof LocalTransactionExecutor<?> localTransactionExecutor) {
            String txId = getTxId(msg);
            // 执行本地事务
            localTransactionExecutor.execute(txId, msg);
            // 记录本地事务记录
            localTransactionExecutor.addLocalTransactionRecord(() -> localTransactionRecordRepository.add(txId));
            return RocketMQLocalTransactionState.COMMIT;

        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        // 正常情况不会调用该方法，业务上一般需要对此方法做监控预警
        // RocketMQ通过这里来回查上面的事务的状态，默认1min回查一次，一共回查15次
        String txId = getTxId(message);
        boolean exist = localTransactionRecordRepository.exist(txId);
        if (exist) {
            return RocketMQLocalTransactionState.COMMIT;
        }

        String topic = getTopic(message);
        log.warn("本地事务消息ID不存在. topic: {}, txId: {}", topic, txId);
        return RocketMQLocalTransactionState.UNKNOWN;
    }

    private String getTxId(Message msg) {
        Object txId = msg.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID);
        if (null == txId) {
            return null;
        }
        return txId.toString();
    }

    private String getTopic(Message msg) {
        Object topic = msg.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TOPIC);
        if (null == topic) {
            return null;
        }
        return topic.toString();
    }
}
